package com.fwmagic.flink.projectcase.twophase;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TwoPhaseCommitDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        env.enableCheckpointing(10000);

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000));

        env.setStateBackend(new FsStateBackend("file:///Users/fangwei/Downloads/chk"));

        //程序异常退出或者人工cancel掉，checkpoint不删除，保留！
        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        DataStreamSource<String> dataStream = env.socketTextStream("localhost", 8888);

        //手动制造异常：用于测试checkpoint数据被正确存储和读取
        dataStream.map(new MapFunction<String, String>() {
            @Override
            public String map(String line) throws Exception {
                if (line.equals("err")) {
                    throw new RuntimeException("程序异常！");
                }
                return line;
            }
        }).print();

        dataStream.print();
        dataStream.addSink(new MyTwoPhaseCommitSink());

        env.execute("TwoPhaseCommitDemo");

    }
}
